package org.example.iotdb.flink.map;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;

import org.apache.iotdb.flink.Event;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.example.iotdb.flink.entity.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Optional;

/**
 * @Auther: kevin
 * @Description:
 * @Company: 上海博般数据技术有限公司
 * @Version: 1.0.0
 * @Date: Created in 11:52 2021/8/20
 * @ProjectName: iotdb-learn
 */
public class MessageToEventMap implements MapFunction<Tuple2<String, Message>, Tuple2<String, Event>> {

    private Logger logger = LoggerFactory.getLogger(this.getClass());

    @Override
    public Tuple2<String, Event> map(Tuple2<String, Message> input) throws Exception {
        try {
            Message message = input.f1;
            String device = "root." + message.getPath();
            List<String> measurements = new ArrayList<>();
            List<TSDataType> types = new ArrayList<>();
            List<Object> values = new ArrayList<>();
            Optional.ofNullable(message.getP())
                    .ifPresent(v -> {
                        measurements.add("P");
                        types.add(TSDataType.DOUBLE);
                        values.add(v);
                    });
            Optional.ofNullable(message.getQ())
                    .ifPresent(v -> {
                        measurements.add("Q");
                        types.add(TSDataType.DOUBLE);
                        values.add(v);
                    });
            Optional.of(message.getI())
                    .ifPresent(v -> {
                        measurements.add("I");
                        types.add(TSDataType.DOUBLE);
                        values.add(v);
                    });
            Optional.of(message.getU())
                    .ifPresent(v -> {
                        measurements.add("U");
                        types.add(TSDataType.DOUBLE);
                        values.add(v);
                    });
            if (!measurements.isEmpty()) {
                Event event = new Event(device, message.getTime(), measurements, types, values);
                return new Tuple2<>(input.f0, event);
            }
        } catch (Exception e) {
            e.printStackTrace();
            logger.error("convert to event failed ! Cause:" + e.getMessage(), e);
        }
        return null;
    }

}
